package com.xzx.spark.core.exercise

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.immutable

/**
 *
 * 需求 2：页面单跳转换率统计
 * 同一个session的，按时间顺序排序，
 * 例如：首页-详情-下单-支付 首页
 * 首页到详情的单条转换率=1/2
 *先算分母，再算分子
 *
 * @author xinzhixuan
 * @version 1.0
 * @date 2021-08-08 2:14 下午
 */
object Exercise04 {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName(getClass.getSimpleName))
    val rdd = sc.textFile("src/main/scala/com/xzx/spark/core/exercise/user_visit_action.txt")
      .map(x => {
        val data = x.split("_")
        UserVisitAction(data(0), data(1).toLong, data(2), data(3).toLong, data(4), data(5), data(6).toLong, data(7).toLong, data(8), data(9), data(10), data(11), data(12).toLong)
      }).cache()
    val pageIdCnt: Map[Long, Int] = rdd.map(x => (x.page_id, 1)).reduceByKey(_ + _).collect().toMap
    rdd.map(x => (x.session_id, x)).groupByKey().mapValues((x: Iterable[UserVisitAction]) => {
      val list = x.toList.sortBy(x => x.date).map(x => x.page_id)
      val tuples: List[(Long, Long)] = list.zip(list.tail)
      tuples.map((_,1))
    }).flatMap(_._2).reduceByKey(_ + _).foreach{
      case ((page1, page2), cnt) =>
        val d: Double = cnt.toDouble / pageIdCnt.getOrElse(page1, 0)
        println(s"页面${page1}跳转到${page2}的单跳转换率为：$d")
    }

    sc.stop()
  }

}

